﻿/**
 * Copyright (C) 2012 - present by Privosoft LLC. 
 * 
 * Please see distribution for license.
 */
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.Data.SqlClient;
using Privosoft.DataFrame.DomainObjects;

namespace Privosoft.DataFrame
{
    public class DbContext : IDisposable
    {

        private readonly SqlConnection cn;
        private readonly string dbSchema;

        SqlBulkCopy bcp;

        public DbContext(SqlConnection cn_) : this(cn_, "dbo") // provided for convenience and existing clients
        {
        }

        public DbContext(SqlConnection cn_, string dbSchema_)
        {
            cn = cn_;
            dbSchema = dbSchema_;

            if(cn.State == ConnectionState.Closed) cn.Open();
            bcp = new SqlBulkCopy(cn);
            bcp.BatchSize = 10000;
            bcp.DestinationTableName = dbSchema + "DataFrameItem";
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(0, 0));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(1, 1));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(2, 2));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(3, 3));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(4, 4));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(5, 5));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(6, 6));
            bcp.ColumnMappings.Add(new SqlBulkCopyColumnMapping(7, 7));
        }

        public void BulkCopy(DataTable dt)
        {
            bcp.WriteToServer(dt);
            dt.Rows.Clear();
        }

        // load summary information about available dataframes
        public ICollection<DataFrameInfo> GetDataFrameInfos(int id = -1)
        {
            var frames = new List<DataFrameInfo>();
            var sql =
                    @"declare @row_count_table TABLE
                    (
	                    DataFrameId int,
	                    RowCt int
                    );

                    insert into @row_count_table 
                    select DataFrameId, max(RowIndex)+1 as RowCt 
                    from @DB_SCHEMA@DataFrameItem dft 
                    @WHERE_ITEM@
                    group by DataFrameId;

                    select df.Id as DataFrameId, df.CreatedTimestamp, df.UserName, dfr.RowCt, df.SchemaId, s.Name as SchemaName 
                    from @DB_SCHEMA@DataFrameInfo df
                    inner join @row_count_table dfr on dfr.DataFrameId = df.Id
                    inner join @DB_SCHEMA@SchemaInfo s on s.Id = df.SchemaId
                    @WHERE@";

            sql = sql.Replace("@DB_SCHEMA@", dbSchema);
            sql = sql.Replace("@WHERE_ITEM@", id >= 0 ? "where dft.DataframeId = " + id : string.Empty);
            sql = sql.Replace("@WHERE@", id >= 0 ? "where df.Id = " + id : string.Empty);

            using (var cmd = new SqlCommand(sql, cn))
            {
                using (var rd = cmd.ExecuteReader())
                {
                    while (rd.Read())
                    {
                        var dfi = new DataFrameInfo
                                    {
                                        Id = rd.GetInt32(0),                                        
                                        CreatedTimestamp = rd.GetDateTime(1),
                                        Username = rd.GetString(2),
                                        RowCount = rd.GetInt32(3),
                                        Schema = new SchemaInfo 
                                                    { 
                                                        Id = rd.GetInt32(4), 
                                                        Name = rd.GetString(5) 
                                                    }
                                    };                       
                        frames.Add(dfi);
                    }
                }
            }

            // join frame info with properties
            try
            {
                var lookup = frames.ToDictionary(df => df.Id);
                sql =
                        @"
                        select t2.DataFrameId, t2.Name, t2.Value
                        from @DB_SCHEMA@DataFrameInfo t1
                        inner join @DB_SCHEMA@DataFrameInfoProps t2 on t1.Id = t2.DataFrameId
                        order by t2.DataFrameId asc";

                sql = sql.Replace("@DB_SCHEMA@", dbSchema);
                using (var cmd = new SqlCommand(sql, cn))
                {
                    using (var rd = cmd.ExecuteReader())
                    {
                        var prevId = -1;
                        DataFrameInfo dfi = null;
                        while (rd.Read())
                        {
                            var did = rd.GetInt32(0);                            
                            var name = rd.GetString(1);
                            var value = rd.GetString(2);

                            if(prevId != did)
                            {
                                prevId = did;
                                if (lookup.TryGetValue(did, out dfi) == false) continue;
                            }
                            if (dfi != null)
                            {
                                if (dfi.Properties == null) dfi.Properties = new Dictionary<string, string>();

                                dfi.Properties.Add(name, value);
                            }
                        }
                    }
                }
            }
            finally
            {
                
            }

            return frames;
        }

        public DataTable GetDataFrameById(int frameId)
        {
            DataTable dt = new DataTable();
            // get columns info
            var sql = @"select T2.Id, T2.Name, T2.DataType 
                from @DB_SCHEMA@DataFrameInfo T1, @DB_SCHEMA@ColumnInfo T2                
                where T1.Id = @frameId and T1.SchemaId = T2.SchemaId and T2.Active=1 order by T2.[Index]";

            sql = sql.Replace("@DB_SCHEMA@", dbSchema);

            using (SqlCommand cm = new SqlCommand(sql, cn))
            {
                var tableColumnInfo = new Dictionary<int, ColumnInfo>();
                #region CONSTRUCT TABLE
                cm.Parameters.Add(new SqlParameter("frameId", frameId));
                SqlDataReader dr = cm.ExecuteReader();
                var idx = 0;
                while (dr.Read())
                {
                    string type = dr.GetString(2).ToLower();
                    if (type == "datetime")
                        type = "DateTime";
                    else
                        type = type.Substring(0, 1).ToUpper() + type.Substring(1);

                    if (type.Equals("Long"))
                        type = "Int64";
                    else if (type.Equals("Double"))
                        type = "Decimal";

                    var tp = System.Type.GetType("System." + type);
                    DataColumn dc = new DataColumn(dr.GetString(1), tp);
                    dt.Columns.Add(dc);

                    tableColumnInfo.Add(dr.GetInt32(0),
                            new ColumnInfo
                            {
                                Id = dr.GetInt32(0),
                                Name = dr.GetString(1),
                                DataType = tp,
                                Index = idx++
                            });
                }
                dr.Close();
                #endregion
                #region FILL ROWS
                int rowIndex = -1;
                DataRow row = null;
                var sw = System.Diagnostics.Stopwatch.StartNew();

                
                sql = @"SELECT T1.[RowIndex], T1.columnId, DoubleValue, StringValue, DateTimeValue, LongValue, BooleanValue
                        FROM @DB_SCHEMA@DataFrameItem T1
                        WHERE T1.DataFrameId = @frameId ORDER BY T1.[RowIndex]";

                sql = sql.Replace("@DB_SCHEMA@", dbSchema);

                cm.CommandText = sql;
                dr = cm.ExecuteReader();
                while (dr.Read())
                {
                    if (rowIndex != dr.GetInt32(0))
                    {
                        if (row != null) dt.Rows.Add(row);
                        row = dt.NewRow();
                        rowIndex = dr.GetInt32(0);
                    }
                    //var allFields = dr.GetSqlValues()
                    if (tableColumnInfo.ContainsKey(dr.GetInt32(1)))
                    {
                        var colInfo = tableColumnInfo[dr.GetInt32(1)];
                        var dataType = colInfo.DataType;
                        if (dataType == typeof(Decimal))
                        {
                            if (!dr.IsDBNull(2)) row[colInfo.Index] = dr.GetDecimal(2);
                        }
                        else if (dataType == typeof(string))
                        {
                            if (!dr.IsDBNull(3)) row[colInfo.Index] = dr.GetString(3);
                        }
                        else if (dataType == typeof(DateTime))
                        {
                            if (!dr.IsDBNull(4)) row[colInfo.Index] = dr.GetDateTime(4);
                        }
                        else if (dataType == typeof(Int64))
                        {
                            if (!dr.IsDBNull(5)) row[colInfo.Index] = dr.GetInt64(5);
                        }
                        else if (dataType == typeof(bool))
                        {
                            if (!dr.IsDBNull(6)) row[colInfo.Index] = dr.GetBoolean(6);
                        }
                    }
                }

                if (row != null) dt.Rows.Add(row);
                dr.Close();
                sw.Stop();
                System.Diagnostics.Debug.WriteLine(string.Format("Data load took: {0:0.0} secs.", sw.ElapsedMilliseconds / 1000.0));
                #endregion
            }

            return dt;
        }
        // get all schema objects
        public ICollection<SchemaInfo> GetSchemas(bool loadDetail)
        {
            var schemas = new List<SchemaInfo>();
            
            var sql = "select id, [name] from @DB_SCHEMA@schemainfo";
            
            sql = sql.Replace("@DB_SCHEMA@", dbSchema);
           
            using (var cmd = new SqlCommand(sql, cn))
            {
                using (var rd = cmd.ExecuteReader())
                {
                    while (rd.Read())
                    {
                        var s = new SchemaInfo
                                {
                                    Id = rd.GetInt32(0),
                                    Name = rd.GetString(1)
                                };
                        schemas.Add(s);
                    }
                }
            }
            if(loadDetail)
                foreach (var s in schemas)
		            LoadSchemaColumns(s);
            
            return schemas;
        }


        public SchemaInfo GetSchemaByName(string schema)
        {
            SchemaInfo s = null;
            var sql = "select ID from @DB_SCHEMA@SchemaInfo where name = @schema";
            sql = sql.Replace("@DB_SCHEMA@", dbSchema);

            using (var cmd = new SqlCommand(sql, cn))
            {
                SqlParameter p = cmd.Parameters.Add("schema", System.Data.SqlDbType.VarChar);
                p.Value = schema;
                object result = cmd.ExecuteScalar();
                if (result != null){

                    s = new SchemaInfo();
                    s.Id = (int)result;
                    s.Name = schema;
                }
            }
            return s;
        }

        // helps with getting the exact values of the dataframe items.
        // this allows one to quickly search for desired summary info.
        // eg find all dataframes with a specific "asof" date
        public DataTable GetDistinctFirstValues(string schemaName, string columnName, ItemDataTypes type, bool mostRecentOnly)
        {
            if(string.IsNullOrEmpty(columnName)) throw new ArgumentNullException("columnName");

            var sql = string.Format(@"with val_cte (Col, T, DataFrameId)
AS
(
	select distinct Col=di.[{2}], T=info.CreatedTimestamp, di.DataFrameId 
	from {0}DataFrameItem di
	inner join {0}ColumnInfo c on c.Name = '{1}' and di.ColumnId = c.Id
	inner join {0}DataFrameInfo info on di.DataFrameId = info.Id
    inner join {0}SchemaInfo sinfo on sinfo.Name = '{4}' and c.SchemaId = sinfo.Id
	where rowindex = 0
)
select a.* from val_cte a
{3}
order by a.Col desc, a.T desc", dbSchema, columnName, type.ToText(), 
                                    mostRecentOnly ? "inner join (select T2=max(T), Col from val_cte group by col) b on a.Col = b.Col and a.T = b.T2" : string.Empty
                                    , schemaName);
            
            using (var cm = new SqlCommand(sql, cn))
            {                
                var da= new SqlDataAdapter(cm);
                var dt = new DataTable();
                da.Fill(dt);
                return dt;
            }
        }

        public int? GetLatestDataFrameInfo(string schemaName)
        {
            var sql = string.Format(@"with cte (id, ts)
as (
select dfi.Id, dfi.CreatedTimestamp 
from {0}SchemaInfo si
inner join {0}DataFrameInfo dfi on si.Id = dfi.SchemaId
where si.Name = '{1}'
)
select id
from cte
where cte.ts = (select max(ts) from cte)
", dbSchema, schemaName);

            using(var cmd = new SqlCommand(sql,cn))
            {
                return (int?)cmd.ExecuteScalar();
            }
        }

        public void LoadSchemaColumns(SchemaInfo s){

            var result = new List<Column>();
            var sql = "select ID, Name, DataType, Required, Active from @DB_SCHEMA@ColumnInfo where SchemaId = @schemaId and Active = 1";
            sql = sql.Replace("@DB_SCHEMA@", dbSchema);

            using (SqlCommand cm = new SqlCommand(sql, cn))
            {
                SqlParameter p = cm.Parameters.Add("schemaId", System.Data.SqlDbType.Int);
                p.Value = s.Id;
                SqlDataReader dr = cm.ExecuteReader();
                while (dr.Read())
                {
                    Column c = new Column();
                    c.Id = dr.GetInt32(0);
                    c.Name = dr.GetString(1);
                    c.DataType = dr.GetString(2);
                    c.IsRequired = dr.GetBoolean(3);
                    c.IsActive = dr.GetBoolean(4);
                    result.Add(c);
                }
                dr.Close();
            }
            s.Columns = result;
        }

        public DataFrameInfo SaveDataFrame(DataFrameInfo df)
        {
            var transaction = cn.BeginTransaction();

            try
            {
                // save dataframe sql command
                var sql = "insert into @DB_SCHEMA@DataFrameInfo(SchemaId, UserName) values (@schemaId, @userName);select scope_identity();";
                sql = sql.Replace("@DB_SCHEMA@", dbSchema);

                using (var cm = new SqlCommand(sql, cn))
                {
                    cm.Transaction = transaction;
                    SqlParameter schemaId = cm.Parameters.Add("schemaId", SqlDbType.Int);
                    SqlParameter userName = cm.Parameters.Add("userName", SqlDbType.VarChar);
                    schemaId.Value = df.Schema.Id;
                    userName.Value = df.Username;
                    df.Id = Int32.Parse(cm.ExecuteScalar().ToString());
                }

                // save dataframe properties sql command
                if (df.Properties != null)
                {
                    sql = @"insert into @DB_SCHEMA@DataFrameInfoProps values (@dataframeId, @name, @value)";
                    sql = sql.Replace("@DB_SCHEMA@", dbSchema);
                    
                    using (var cm = new SqlCommand(sql, cn))
                    {
                        cm.Transaction = transaction;
                        SqlParameter pid = cm.Parameters.Add("dataframeId", SqlDbType.Int);
                        SqlParameter pname = cm.Parameters.Add("name", SqlDbType.VarChar);
                        SqlParameter pvalue = cm.Parameters.Add("value", SqlDbType.VarChar);

                        foreach(var kv in df.Properties)
                        {
                            if (kv.Value == null) continue;
                            pid.Value = df.Id;
                            pname.Value = kv.Key;
                            pvalue.Value = kv.Value;
                            cm.ExecuteNonQuery();
                        }
                    }
                }

                transaction.Commit();
            }
            catch(Exception e)
            {
                System.Diagnostics.Debug.WriteLine(e.StackTrace);
                transaction.Rollback();
            }
            finally
            {
                ////cn.Close();
            }

            return df;
        }

        public void Dispose()
        {
            cn.Close();
        }

        private class ColumnInfo
        {
            public int Id { get; set; }
            public string Name { get; set; }
            public Type DataType { get; set; }
            public int Index { get; set; }
        }

    }
}
